package com.relin.websocket;

import static org.jboss.netty.handler.codec.http.HttpHeaders.*;  
import static org.jboss.netty.handler.codec.http.HttpHeaders.Names.*;  
import static org.jboss.netty.handler.codec.http.HttpMethod.*;  
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.*;  
import static org.jboss.netty.handler.codec.http.HttpVersion.*;  

import java.util.HashMap;
import java.util.Map;

import org.jboss.netty.buffer.ChannelBuffer;  
import org.jboss.netty.buffer.ChannelBuffers;  
import org.jboss.netty.channel.ChannelFuture;  
import org.jboss.netty.channel.ChannelFutureListener;  
import org.jboss.netty.channel.ChannelHandlerContext;  
import org.jboss.netty.channel.ExceptionEvent;  
import org.jboss.netty.channel.MessageEvent;  
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;  
import org.jboss.netty.handler.codec.http.DefaultHttpResponse;  
import org.jboss.netty.handler.codec.http.HttpHeaders;  
import org.jboss.netty.handler.codec.http.HttpRequest;  
import org.jboss.netty.handler.codec.http.HttpResponse;  
import org.jboss.netty.handler.codec.http.websocketx.CloseWebSocketFrame;  
import org.jboss.netty.handler.codec.http.websocketx.PingWebSocketFrame;  
import org.jboss.netty.handler.codec.http.websocketx.PongWebSocketFrame;  
import org.jboss.netty.handler.codec.http.websocketx.TextWebSocketFrame;  
import org.jboss.netty.handler.codec.http.websocketx.WebSocketFrame;  
import org.jboss.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;  
import org.jboss.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;  
import org.jboss.netty.logging.InternalLogger;  
import org.jboss.netty.logging.InternalLoggerFactory;  
import org.jboss.netty.util.CharsetUtil;  

import redis.clients.jedis.Jedis;

import com.relin.websocket.util.JedisUtils;
  
public class WebSocketServerHandler extends SimpleChannelUpstreamHandler  
{  
    private static final InternalLogger logger = InternalLoggerFactory  
            .getInstance(WebSocketServerHandler.class);  
  
    private static final String WEBSOCKET_PATH = "/websocket";  
    public static final String KEY = "aptoto";
    
    public static Map<ChannelHandlerContext, MessageEvent> map;
    
    private Jedis jedis;
    
    private WebSocketServerHandshaker handshaker;  
    
    @Override  
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)  
            throws Exception {  
        // 处理接受消息  
    	if(map==null){
    		map= new HashMap<ChannelHandlerContext, MessageEvent>(); 
    	}
    	map.put(ctx, e);
    	if(jedis==null){    		
    		jedis = JedisUtils.getJedis();
    	}    	
        Object msg = e.getMessage();  
        if (msg instanceof HttpRequest) {  
            handleHttpRequest(ctx, (HttpRequest) msg);  
        } else if (msg instanceof WebSocketFrame) {  
            handleWebSocketFrame(ctx, (WebSocketFrame) msg);  
        }  
    }  
  
    @Override  
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)  
            throws Exception {  
        // 处理异常情况  
        e.getCause().printStackTrace();  
        e.getChannel().close();  
    }  
  
    private void handleHttpRequest(ChannelHandlerContext ctx, HttpRequest req)  
            throws Exception {  
        // 只接受 HTTP GET 请求  
        if (req.getMethod() != GET) {  
            sendHttpResponse(ctx, req, new DefaultHttpResponse(HTTP_1_1,  
                    FORBIDDEN));  
            return;  
        }  
  
        // Websocket handshake 
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(  
                getWebSocketLocation(req), null, false);  
        handshaker = wsFactory.newHandshaker(req);  
        if (handshaker == null) {  
            wsFactory.sendUnsupportedWebSocketVersionResponse(ctx.getChannel());  
        } else {  
            handshaker.handshake(ctx.getChannel(), req).addListener(  
                    WebSocketServerHandshaker.HANDSHAKE_LISTENER);  
        }  
    }  
  
    private void handleWebSocketFrame(ChannelHandlerContext ctx,  
            WebSocketFrame frame) {  
        // Websocket 握手结束  
        if (frame instanceof CloseWebSocketFrame) {  
            handshaker.close(ctx.getChannel(), (CloseWebSocketFrame) frame);  
            return;  
        } else if (frame instanceof PingWebSocketFrame) {  
            ctx.getChannel().write(new PongWebSocketFrame(frame.getBinaryData()));  
            return;  
        } else if (!(frame instanceof TextWebSocketFrame)) {  
            throw new UnsupportedOperationException(String.format("%s frame types not supported",  
                    frame.getClass().getName()));  
        }  
  
        // 处理接受到的数据（转成大写）并返回  
        
        String request = ((TextWebSocketFrame) frame).getText();
        jedis.publish(KEY, request);
        if (logger.isDebugEnabled()) {  
            logger.debug(String.format("Channel %s received %s", ctx.getChannel().getId(), request));  
        }  
        System.out.println(new TextWebSocketFrame(request.toUpperCase()));
        /*ctx.getChannel().write(new TextWebSocketFrame(request.toUpperCase())); */ 
    }  
  
    private static void sendHttpResponse(ChannelHandlerContext ctx,  
            HttpRequest req, HttpResponse res) {  
        // 返回 HTTP 错误页面  
        if (res.getStatus().getCode() != 200) {  
            res.setContent(ChannelBuffers.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8));  
            setContentLength(res, res.getContent().readableBytes());  
        }  
  
        // 发送返回信息并关闭连接  
        ChannelFuture f = ctx.getChannel().write(res);  
        if (!isKeepAlive(req) || res.getStatus().getCode() != 200) {  
            f.addListener(ChannelFutureListener.CLOSE);  
        }  
    }  
  
    private static String getWebSocketLocation(HttpRequest req) {  
        return "ws://" + req.getHeader(HttpHeaders.Names.HOST) + WEBSOCKET_PATH;  
    }  
}  
